package javaVersion.learn.batchProcess;

import javaVersion.pojo.Person2;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapPartitionOperator;
import org.apache.flink.util.Collector;
import java.util.ArrayList;

/**
 * @author Wincoma
 * @date 2020/7/15 15:15
 * @description
 */
public class Pro3_OperatorTest {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        ArrayList<String> list = new ArrayList<>();
        ArrayList<Person2> list2 = new ArrayList<>();
        list.add("1,张三");
        list.add("2,李四");
        list.add("3,王五");
        list.add("4,赵六");
        DataSource<String> data = env.fromCollection(list);

        //map
//        data.map(s -> {
//            String[] split = s.split(",");
//            Person2 p2 = new Person2();
//            p2.setId(Integer.parseInt(split[0]));
//            p2.setName(split[1]);
//            return p2;
//        }).print();

        //foreachPartition
        MapPartitionOperator<String, Person2> mapData = data.mapPartition(new MapPartitionFunction<String, Person2>() {
            @Override
            public void mapPartition(Iterable<String> values, Collector<Person2> out) throws Exception {
                if (values != null) {
                    for (String value : values) {
                        String[] split = value.split(",");
                        Person2 person2 = new Person2();
                        person2.setId(Integer.parseInt(split[0]));
                        person2.setName(split[1]);
                        System.out.println(person2);
                    }
                }
            }
        });
        mapData.print();

    }
}
